package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.bean.KeywordBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3. 从kafka的 主题中读取数据
        String topic = "edu_dwd_traffic_page";
        String group = "edu_dwd_traffic_keyword_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");

        // TODO 4. 转为json数据
        SingleOutputStreamOperator<JSONObject> etlJsonObjectDS =
                kafkaDS
                        .map(JSON::parseObject)
                        .filter(in -> "keyword".equals(in.getJSONObject("page").getString("item_type")));
//        etlJsonObjectDS.print("+++++++++++++++");

        // TODO 5. 设置水位线
        SingleOutputStreamOperator<JSONObject> watermarkDS = etlJsonObjectDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        })
        );

        // TODO 6. 以keyword 分组，开窗，聚合
        SingleOutputStreamOperator<KeywordBean> reduceDS = watermarkDS
                .keyBy(in -> in.getJSONObject("page").getString("item"))
                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
                .aggregate(new AggregateFunction<JSONObject, Long, Long>() {
                               @Override
                               public Long createAccumulator() {
                                   return 0L;
                               }

                               @Override
                               public Long add(JSONObject value, Long accumulator) {
                                   return accumulator + 1L;
                               }

                               @Override
                               public Long getResult(Long accumulator) {
                                   return accumulator;
                               }

                               @Override
                               public Long merge(Long a, Long b) {
                                   return null;
                               }
                           },
                        new WindowFunction<Long, KeywordBean, String, TimeWindow>() {
                            @Override
                            public void apply(String keyword, TimeWindow window, Iterable<Long> input, Collector<KeywordBean> out) throws Exception {
                                String sst = DateUtil.toYmdHms(window.getStart());
                                String edt = DateUtil.toYmdHms(window.getEnd());
                                out.collect(new KeywordBean(
                                        sst,
                                        edt,
                                        keyword,
                                        input.iterator().next(),
                                        System.currentTimeMillis()
                                ));
                            }
                        });

        reduceDS.print("》》》》》》》》》");

        // TODO 6. 写入clickhouse中
        reduceDS.addSink(ClickhouseUtil.getSinkFunction("insert into dws_traffic_keyword_page_view_window values(?,?,?,?,?)"));

        env.execute();
    }
}
